// worker_thread.cpp


#include <csignal>
#include <cstdio>
#include <cstdlib>

#include <chrono>
#include <condition_variable>
#include <mutex>
#include <sstream>
#include <string>
#include <thread>
#include <vector>


namespace blogs
{


using std::chrono::milliseconds;
using std::condition_variable;
using std::mutex;
using std::ostringstream;
using std::string;
using std::thread;
using std::unique_lock;
using std::vector;


string this_thread_id()
{
    ostringstream oss;
    oss << std::this_thread::get_id();
    return oss.str();
}


class Request
{

  private:

    string
    __name__;


    int
    __no__;


  public:

    Request(string name = "", int no = 0)
        : __name__(name), __no__(no)
    {
    }


    void execute()
    {
        string msg;
        msg += this_thread_id();
        msg += " executes [Request from ";
        msg += __name__;
        msg += " No. ";
        msg += std::to_string(__no__);
        msg += " ]\n";
        std::printf("%s", msg.c_str());
        std::this_thread::sleep_for(milliseconds(std::rand() % 1000));
    }

};


class Worker;


class Channel
{

  private:

    static
    int const
    __MAX_REQUEST_NO__ = 10;


    vector<thread>
    __threadpool__;


    vector<Request>
    __requestqueue__;


    int
    __head__, __tail__, __count__;


    mutex
    __mtx__;


    condition_variable
    __cv__;


  public:

    Channel(int wrkrnum)
        : __requestqueue__(__MAX_REQUEST_NO__+1), __head__(0), __tail__(0), __count__(0), __threadpool__(wrkrnum)
    {
    }


    vector<thread> &
    start_workers();


    void
    put_request(Request req)
    {
        unique_lock<mutex> lk(__mtx__);
        __cv__.wait(lk,
                    [this] {
                        bool full = __count__ >= __MAX_REQUEST_NO__;
                        if (full)
                            std::printf("Request queue is full. Waiting...\n");
                        return !full;
                    });
        __requestqueue__[__tail__] = req;
        __tail__ = (__tail__ + 1) % __MAX_REQUEST_NO__;
        ++__count__;
        __cv__.notify_all();
    }


    Request
    take_request()
    {
        unique_lock<mutex> lk(__mtx__);
        __cv__.wait(lk,
                    [this] {
                        bool empty = __count__ <= 0;
                        if (empty)
                            std::printf("No requests in queue. Waiting...\n");
                        return !empty;
                    });
        Request req = __requestqueue__[__head__];
        __head__ = (__head__ + 1) % __MAX_REQUEST_NO__;
        --__count__;
        __cv__.notify_all();
        return req;
    }

};


class Worker
{

  private:

    string
    __name__;


    Channel &
    __channel__;


  public:

    Worker(string name, Channel & channel)
        : __name__(name), __channel__(channel)
    {
    }


    void run()
    {
        string msg;
        msg += __name__;
        msg += "'s id: ";
        msg += this_thread_id();
        std::printf("%s\n", msg.c_str());
        while (true) {
            Request req = __channel__.take_request();
            req.execute();
        }
    }

};


class Client
{

  private:

    string
    __name__;


    Channel &
    __channel__;


  public:

    Client(string name, Channel & channel)
        : __name__(name), __channel__(channel)
    {
    }


    void run()
    {
        for (int i = 1; true; ++i) {
            __channel__.put_request(Request(__name__, i));
            std::this_thread::sleep_for(milliseconds(std::rand() % 600));
        }
    }

};


vector<thread> &
Channel::start_workers()
{
    for (int i = 0; i < __threadpool__.size(); ++i) {
        string name("Worker-");
        name += std::to_string(i);
        __threadpool__[i] = thread(&Worker::run, Worker(name, *this));
    }
    return __threadpool__;
}


void
bye(int sig = SIGTERM)
{
    std::printf("\nBye Worker Thread...\n\n");
    std::exit(sig);
}


} // end of namespace blogs


int
main(int argc, char * argv[])
{
    using std::chrono::milliseconds;
    using std::thread;
    using std::vector;
    using blogs::Channel;
    using blogs::Client;

    std::signal(SIGINT, blogs::bye);
    std::srand(std::time(0));

    Channel channel(5);
    vector<thread> & workers = channel.start_workers();

    Client alice("Alice", channel);
    Client bobby("Bobby", channel);
    Client chris("Chris", channel);

    thread ct1(&Client::run, &alice);
    thread ct2(&Client::run, &bobby);
    thread ct3(&Client::run, &chris);

    std::this_thread::sleep_for(milliseconds(60));

    for (int i = 0; i < workers.size(); ++i)
        workers[i].join();
    ct1.join();
    ct2.join();
    ct3.join();

    blogs::bye();
    return 0;
}

